package analysis.newuser;

import analysis.bean.BrowserDimension;
import analysis.bean.DateDemension;
import analysis.bean.KPIDemension;
import analysis.bean.PlatformDemension;
import analysis.bean.key.StatsCommon;
import analysis.bean.key.StatsUser;
import analysis.bean.value.MapWritableValue;
import analysis.bean.value.TimeOutputValue;
import common.Constants;
import common.DateEnum;
import common.KPIType;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

public class NewUser {
    public static class M extends Mapper<LongWritable, Text, StatsUser, TimeOutputValue> {

        public StatsUser k = new StatsUser();
        public TimeOutputValue v = new TimeOutputValue();

        public KPIDemension userKPI = new KPIDemension(KPIType.NEW_USER.kpiName);
        public KPIDemension browserKPI = new KPIDemension(KPIType.BROWSER_NEW_USER.kpiName);

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            if (line.isEmpty())
                return;
            String fields[] = line.split("\\u0001");
            String en = fields[2];
            if (!en.isEmpty() && en.equals(Constants.EventEnum.LANUCH)) {
                String serverTime = fields[1];
                String platform = fields[13];
                String uuid = fields[4];
                String browserName = fields[24];
                String browserVersion = fields[25];
                if (uuid.isEmpty() || uuid.equals("null") || serverTime.isEmpty())
                    return;
                long time = Long.valueOf(serverTime);

                v.setUuid(uuid);

                PlatformDemension platformDemension = new PlatformDemension(platform);
                DateDemension dateDimension = DateDemension.buildDate(time, DateEnum.DAY);

                StatsCommon statsCommon = new StatsCommon();
                statsCommon.setDateDemension(dateDimension);
                statsCommon.setPlatformDemension(platformDemension);
                statsCommon.setKpiDemension(userKPI);

                BrowserDimension emptyBrowserDimension = new BrowserDimension("", "");

                k.setStatsCommon(statsCommon);
                k.setBrowserDimension(emptyBrowserDimension);
                context.write(k, v);

                statsCommon.setKpiDemension(browserKPI);
                BrowserDimension browserDimension = new BrowserDimension(browserName, browserVersion);
                k.setBrowserDimension(browserDimension);
                k.setStatsCommon(statsCommon);
                context.write(k, v);
            }
        }
    }

    public static class R extends Reducer<StatsUser, TimeOutputValue, StatsUser, MapWritableValue> {
        public MapWritable map = new MapWritable();

        @Override
        protected void reduce(StatsUser key, Iterable<TimeOutputValue> values, Context context) throws IOException, InterruptedException {
            Set set = new HashSet();

            for(TimeOutputValue user : values)
            {
                set.add(user.getUuid());
            }
            MapWritableValue v = new MapWritableValue();
            v.setKpi(KPIType.valueOfKpiName(key.getStatsCommon().getKpiDemension().getKPIName()));
            this.map.put(new IntWritable(1),new IntWritable(set.size()));
            v.setValue(map);
            context.write(key,v);
            set.clear();

        }
    }
}
